package com.zcy;

import com.espertech.esper.common.client.EventBean;
import com.espertech.esper.runtime.client.EPEventService;
import com.zcy.event.MyEvent;
import com.zcy.util.EsperBuilder;
import com.zcy.util.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.util.concurrent.TimeUnit;

/**
 * 6.2.3. Selecting From Named Windows
 */
@Slf4j
public class NamedWindow extends TimeUtils {

    // 创建窗口 并向其中添加数据  查看插入流和删除流
    private static final String _NamedWindow = "create window MyEventWindow#time(2 sec) as select amount, num, timestamp from MyEvent;";
    private static final String _InsertIntoWindow = "insert into MyEventWindow(amount, num, timestamp) select amount, num, timestamp from MyEvent;";
    private static final String _Normal = "@name('normal-statement') select * from MyEventWindow;";
    private static final String _Time = "@name('rstream-statement') select rstream * from MyEventWindow";

    /**
     * 监听插入流
     */
    @Test
    public void lengthWindowTest() {
        EsperBuilder esperBuilder = new EsperBuilder(_NamedWindow + "\n " + _InsertIntoWindow + "\n " + _Normal, new Class[]{MyEvent.class});

        esperBuilder.startCompile();

        EPEventService epEventService = esperBuilder.startDeploy((eventBeans, eventBeans1, epStatement, epRuntime) -> {
            for (int i = 0; i < eventBeans.length; i++) {
                EventBean eventBean = eventBeans[i];
                log.info("amount is : {}", eventBean.get("amount"));
            }
        });

        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(10).build(), MyEvent.class.getSimpleName());

        sleep(TimeUnit.SECONDS, 2L);

        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("002").name("002-name").num(5).build(), MyEvent.class.getSimpleName());

        sleep(TimeUnit.SECONDS, 40L);
    }


    /**
     * 监听删除流
     */
    @Test
    public void rStreamNamedWindowTest() {
        EsperBuilder esperBuilder = new EsperBuilder(_NamedWindow + "\n " + _InsertIntoWindow + "\n " + _Time, new Class[]{MyEvent.class});

        esperBuilder.startCompile();

        EPEventService epEventService = esperBuilder.startDeploy((eventBeans, eventBeans1, epStatement, epRuntime) -> {
            for (int i = 0; i < eventBeans.length; i++) {
                EventBean eventBean = eventBeans[i];
                log.info("amount is : {}", eventBean.get("amount"));
            }
        });

        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(10).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 3L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("002").name("002-name").num(5).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 40L);
    }


}
